package _01_quick_start._8_2_2_time_based_processing

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl._

import scala.concurrent.duration._

object Main extends App {

  implicit val system       = ActorSystem("QuickStart")
  implicit val materializer = ActorMaterializer()
  implicit val ec           = system.dispatcher

  // source
  val source: Source[Int, NotUsed] = Source(1 to 100)

  // flow
  val flow = Flow[Int]
             .scan(BigInt(1))((acc, next) => acc * next)

  // 使用`throttle` 协调器(combinator)，按1秒一个元素的处理速度
  val throttle = flow
                 .zipWith(Source(0 to 100))((num, idx) => s"$idx! = $num")
                 .throttle(1, 1.second, 1, ThrottleMode.shaping)

  val done = source.via(throttle).runForeach(println)

  done.onComplete(_ => system.terminate())
}
